package cn.myframe.config;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;

/**
 * @Author: ynz
 * @Date: 2019/1/28/028 13:29
 * @Version 1.0
 */
@Component
@ConditionalOnBean(FileSystem.class)

public class HadoopTemplate {

    @Autowired
    private FileSystem fileSystem;

    @Value("${hadoop.name-node}")
    private String nameNode;

    @Value("${hadoop.namespace:/}")
    private String nameSpace;

    @PostConstruct
    public void init(){
        existDir(nameSpace,true);
    }

    public void uploadFile(String srcFile){
        copyFileToHDFS(false,true,srcFile,nameSpace);
    }

    public void uploadFile(boolean del,String srcFile){
        copyFileToHDFS(del,true,srcFile,nameSpace);
    }

    public void uploadFile(String srcFile,String destPath){
        copyFileToHDFS(false,true,srcFile,destPath);
    }

    public void uploadFile(boolean del,String srcFile,String destPath){
        copyFileToHDFS(del,true,srcFile,destPath);
    }

    public void delFile(String fileName){
        rmdir(nameSpace,fileName) ;
    }

    public Boolean delDir(String path){
        return rmdir(getRelativeHdfsUrl(path) ,null) ;
    }

    public void download(String fileName,String savePath){
        getFile(getRelativeHdfsUrl(fileName),savePath);
    }


    /**
     * 重命名
     * @param oldPath 旧路径
     * @param newPath 新路径
     * @return
     */
    public boolean rename(String oldPath,String newPath) {
        try {
            if(existFile(oldPath)) {
                return fileSystem.rename(new Path(getRelativeHdfsUrl(oldPath)), new Path(getRelativeHdfsUrl(newPath)));
            }else {
                System.out.println("重命名的对象不存在："+oldPath);
            }
        }
        catch (IOException e) {
            e.printStackTrace();
        }
        return  false;
    }

    /**
     * 创建目录
     * @param filePath
     * @param create
     * @return
     */
    public boolean existDir(String filePath, boolean create){
        boolean flag = false;
        if(StringUtils.isEmpty(filePath)){
            return false;
        }
        try{
            Path path = new Path(filePath);
            if (create){
                if (!fileSystem.exists(path)){
                    fileSystem.mkdirs(path);
                }
            }
            if (fileSystem.isDirectory(path)){
                flag = true;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return flag;
    }

    public FileStatus[] listFile(String filePath ){
        try {

            FileStatus[] listStatus = fileSystem.listStatus(new Path(filePath));
            String name = "";
            for (FileStatus fileStatus : listStatus) {
                name = fileStatus.getPath().toString()  ;
                System.out.println("文件夹或文件夹：" + name.split(filePath)[1]);
                //digui(fileStatus, filePath);
            }
            fileSystem.close();   //关闭
            return listStatus;
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return new FileStatus[0] ;
    }

    /**
     * 递归遍历文件夹
     * @param fileStatus
     * @param filePath
     */
    public  void digui (FileStatus fileStatus,String filePath){
        String name = "";
        if (fileStatus.isDirectory()) {
            name = fileStatus.getPath().toString()  ;
            System.out.println("文件夹：" + name.split(filePath)[1]);
            try {
                FileStatus[] listStatus = fileSystem.listStatus(new Path(name.split(filePath)[1]));
                if (listStatus.length != 0) {
                    for (FileStatus fileStatus2 : listStatus) {
                        digui(fileStatus2, filePath);
                    }
                }
            } catch (FileNotFoundException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (IllegalArgumentException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }


        } else {
            System.out.println("文件:" + (fileStatus.getPath() + "").split(filePath)[1]);
        }

    }

    public boolean existFile(String filePath ){
        boolean flag = false;
        if(StringUtils.isEmpty(filePath)){
            throw new IllegalArgumentException("filePath不能为空");
        }
        try{
            Path path = new Path(getRelativeHdfsUrl(filePath));
            if (fileSystem.isFile(path)){
                flag = true;
            }
        }catch (Exception e){
            e.printStackTrace();
        }
        return flag;
    }


    /**
     * HDFS读取文件流
     * @param srcPath
     * @return
     */
    public InputStream readStream(String srcPath){
        Path path = new Path(getRelativeHdfsUrl(srcPath));
        FSDataInputStream fdis = null;
        try {
            fdis = fileSystem.open(path);
            InputStream str = fdis.getWrappedStream();
            fdis.close();
            return str;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }


    /**
     * HDFS写入文件流
     * @param srcPath
     * @return
     */
    public boolean writeStream(InputStream stream ,String srcPath){
        try {
            Path path = new Path(getRelativeHdfsUrl(srcPath));
            FSDataOutputStream fdos = fileSystem.create(path);
            byte[] tempbytes = new byte[1024000];
            int byteread = 0;
            while ((byteread = stream.read(tempbytes)) != -1) {

                fdos.write(tempbytes);
            }
            fdos.close();
            return true;
        } catch (IOException e) {
            e.printStackTrace();
        }

        return false;
    }


    /**
     * 文件上传至 HDFS
     * @param delSrc       指是否删除源文件，true为删除，默认为false
     * @param overwrite
     * @param srcFile      源文件，上传文件路径
     * @param destPath     hdfs的目的路径
     */
    public  void copyFileToHDFS(boolean delSrc, boolean overwrite,String srcFile,String destPath) {
        // 源文件路径是Linux下的路径，如果在 windows 下测试，需要改写为Windows下的路径，比如D://hadoop/djt/weibo.txt
        Path srcPath = new Path(srcFile);

        // 目的路径
        if(StringUtils.isNotBlank(nameNode)){
            destPath = nameNode + destPath;
        }
        Path dstPath = new Path(destPath);
        // 实现文件上传
        try {
            // 获取FileSystem对象
            //  fileSystem.copyFromLocalFile(srcPath, dstPath);
            fileSystem.copyFromLocalFile(delSrc,overwrite,srcPath, dstPath);
            //释放资源
            //    fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 删除文件或者文件目录
     *
     * @param path
     */
    private Boolean rmdir(String path,String fileName) {
        try {
            // 返回FileSystem对象
            if(StringUtils.isNotBlank(nameNode)){
                path = getRelativeHdfsUrl(path);
            }
            if(StringUtils.isNotBlank(fileName)){
                path =  path + "/" +fileName;
            }
            // 删除文件或者文件目录  delete(Path f) 此方法已经弃用
            return   fileSystem.delete(new Path(path),true);
        } catch (IllegalArgumentException | IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 从 HDFS 下载文件
     *
     * @param hdfsFile
     * @param destPath 文件下载后,存放地址
     */
    public void getFile(String hdfsFile,String destPath) {
        // 源文件路径
        if(StringUtils.isNotBlank(nameNode)){
            hdfsFile = nameNode + hdfsFile;
        }
        Path hdfsPath = new Path(hdfsFile);
        Path dstPath = new Path(destPath);
        try {
            // 下载hdfs上的文件
            fileSystem.copyToLocalFile(hdfsPath, dstPath);
            // 释放资源
            // fs.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public String getNameSpace(){
        return nameSpace;
    }
    public String getHdfsUrl(){
        return nameNode+nameSpace;
    }
    public String getRelativeHdfsUrl(String spath){
        return  nameSpace+"/"+spath;
    }

}

